
#undef LOGGER
#define NDEBUG

#include <boost/scoped_ptr.hpp>
#include <xsocket/active_reactor.hpp>

// -*-mode:c++; coding:utf-8-*-

#ifndef _HANDLER_STATUS_
#define _HANDLER_STATUS_

struct handler_status
{
	unsigned int m_left_in_message;
	unsigned int m_left_out_message;

	handler_status()
		: m_left_in_message(0)
		, m_left_out_message(0) {}
};

#endif	// _HANDLER_STATUS_

// -*-mode:c++; coding:utf-8-*-


#ifndef _CLIENT_MESSAGE_HANDLER_HPP_
#define _CLIENT_MESSAGE_HANDLER_HPP_

#include <utility>
#include <sstream>
#include <string>
#include <boost/shared_ptr.hpp>
#include <boost/thread/condition.hpp>
#include <boost/thread/thread.hpp>

#include <xthread/thread.hpp>
#include <appframe/circular_array_queue.hpp>
#include <xsocket/bytebuffer.hpp>

#include <stringable.hpp>

template<typename T,
	 typename Mutex,
	 typename Condition,
	 std::size_t slot_number>
class basic_message_queue
{
public:
	typedef Mutex mutex_type;
	typedef Condition condition_type;
	typedef std::size_t size_type;
	typedef T value_type;

	enum { CAPACITY = slot_number };

public:
	basic_message_queue() 
		: m_head(0)
		, m_tail(0) {}

	size_type capacity() const {
		return CAPACITY;
	}

	bool full() const {
		typename mutex_type::scoped_lock lock(m_mutex);
		return full_i();
	}

	bool empty() const {
		typename mutex_type::scoped_lock lock(m_mutex);
		return empty_i();
	}

	size_type size() const {
		typename mutex_type::scoped_lock lock(m_mutex);
		return size_i();
	}

	void clear() {
		typename mutex_type::scoped_lock lock(m_mutex);
		clear_i();
	}

	// const value_type &front() const {
	// 	typename mutex_type::scoped_lock lock(m_mutex);
	// 	return front_i();
	// }

	// value_type &front() {
	// 	typename mutex_type::scoped_lock lock(m_mutex);
	// 	return front_i();
	// }

	// const value_type &back() const {
	// 	typename mutex_type::scoped_lock lock(m_mutex);
	// 	return back_i();
	// }

	// value_type &back() {
	// 	typename mutex_type::scoped_lock lock(m_mutex);
	// 	return back_i();
	// }

	bool push_back(const value_type &message) {
		typename mutex_type::scoped_lock lock(m_mutex);
		const bool is_not_full = (! full_i());
		if(is_not_full)
		{
			push_back_i(message);
			m_empty_monitor.notify_one();
		}
		return is_not_full;
	}

	// bool push_front(const value_type &message) {
	// 	typename mutex_type::scoped_lock lock(m_mutex);
	// 	const bool is_not_full = (! full_i());
	// 	if(is_not_full)
	// 	{
	// 		push_front_i(message);
	// 		m_empty_monitor.notify_one();
	// 	}
	// 	return is_not_full;
	// }

	bool block_push_back(const value_type &message) {
		typename mutex_type::scoped_lock lock(m_mutex);
		for(;full_i();)
		{
			m_full_monitor.wait(m_mutex);
		}

		push_back_i(message);
		m_empty_monitor.notify_one();
		return true;
	}

	template<typename Duration>
	bool timed_push_back(const value_type &message,
			     Duration duration) {
		typename mutex_type::scoped_lock lock(m_mutex);
		if(full_i())
		{
			m_full_monitor.wait(m_mutex, duration);
		}
		const bool is_not_full = (! full_i());
		if(is_not_full)
		{
			push_back_i(message);
			m_empty_monitor.notify_one();
		}
		return is_not_full;
	}

	bool pop_front(value_type &message) {
		typename mutex_type::scoped_lock lock(m_mutex);
		const bool is_not_empty = (! empty_i());
		if(is_not_empty)
		{
			pop_front_i(message);
			m_full_monitor.notify_one();
		}
		return is_not_empty;
	}

	bool block_pop_front(value_type &message) {
		typename mutex_type::scoped_lock lock(m_mutex);
		for(;empty_i();)
		{
			m_empty_monitor.wait(m_mutex);
		}
		pop_front_i(message);
		m_full_monitor.notify_one();
		return true;
	}

	template<typename Duration>
	bool timed_pop_front(value_type &message,
			     Duration duration) {
		typename mutex_type::scoped_lock lock(m_mutex);
		if(empty_i())
		{
			m_empty_monitor.wait(m_mutex, duration);
		}
		const bool is_not_empty = (! empty_i());
		if(is_not_empty)
		{
			pop_front_i(message);
			m_full_monitor.notify_one();
		}
		return is_not_empty;
	}

	// push/pop for circular queue, alway push into back and pop
	// from front

	bool push(const value_type &message) {
		return push_back(message);
	}

	bool pop(value_type &message) {
		return pop_front(message);
	}

	bool block_push(const value_type &message) {
		return block_push_back(message);
	}

	bool block_pop(value_type &message) {
		return block_pop_front(message);
	}

	bool timed_push(const value_type &message) {
		return timed_push_back(message);
	}

	bool timed_pop(value_type &message) {
		return timed_pop_front(message);
	}

protected:
	bool full_i() const {
		return size_i() == (CAPACITY - 1);
	}
	
	bool empty_i() const {
		return m_head == m_tail;
	}

	size_type size_i() const {
		return (CAPACITY - m_head + m_tail) % CAPACITY;
	}

	void clear_i() {
		m_head = 0;
		m_tail = 0;
	}

	size_type index_back_i() {
		return backward_positon_i(m_tail);
	}
	
	size_type index_front_i() {
		return m_head;
	}

	// helper for access/push/pop into circular queue
	//
	// m_tail is pointed to index[last element] + 1
	// m_head is pointed to index[first element]

	const value_type &back_i() const {
		return m_storage[backward_positon_i(m_tail)];
	}

	value_type &back_i() {
		return m_storage[backward_positon_i(m_tail)];
	}

	const value_type & front_i() const {
		return m_storage[m_head];
	}

	value_type & front_i() {
		return m_storage[m_head];
	}

	void push_back_i(const value_type &message) {
		m_storage[m_tail] = message;
		m_tail = forward_positon_i(m_tail);
	}

	void push_front_i(const value_type &message) {
		m_head = backward_positon_i(m_head);
		m_storage[m_head] = message;
	}

	void pop_back_i(value_type &message) {
		m_tail = backward_positon_i(m_tail);
		message = m_storage[m_tail];
	}

	void pop_front_i(value_type &message) {
		message = m_storage[m_head];
		m_head = forward_positon_i(m_head);
	}

	// helper for manipulate position of circular queue

	inline
	size_type forward_positon_i(const size_type &position) const {
		return (position + 1) % CAPACITY;
	}

	inline
	size_type backward_positon_i(const size_type &position) const {
		return (CAPACITY + position - 1) % CAPACITY;
	}
private:
	value_type m_storage[CAPACITY];

	size_type m_head;
	size_type m_tail;

	mutex_type m_mutex;
	condition_type m_full_monitor;
	condition_type m_empty_monitor;
};


typedef std::pair<ConnectionInfo*, ByteBuffer*> MessageType;
typedef basic_message_queue<MessageType,
			    boost::mutex,
			    boost::condition,
			    1024 * 1024> MessageQueue;

MessageQueue in_queue;
MessageQueue out_queue;

// struct client_message 
// {
// 	uint32_t m_connection_id;
// 	Header *m_packet;

// 	client_message(uint32_t p_id, Header *p_packet)
// 		: m_connection_id(p_id)
// 		, m_packet(p_packet) {		
// 	}
		
// };

class ClientMessageHandler : public Thread
{
public:

public:

	ClientMessageHandler (unsigned int id,
			      handler_status &status);

	virtual ~ClientMessageHandler () {}

        virtual void run ();

protected:
	void do_read(const MessageType &p_msg);
	void do_write(const MessageType &p_msg);
	void do_scan_open(const MessageType &p_msg);
	void do_scan_next(const MessageType &p_msg);
	void do_scan_close(const MessageType &p_msg);
	void do_unexpected(const MessageType &p_msg);
	void do_list_tablet(const MessageType &p_msg);

private:
	handler_status &m_status;

	// for debug & profile
	unsigned int _id;
	unsigned int _input_count;
	unsigned int _output_count;
};

#endif	// _CLIENT_MESSAGE_HANDLER_HPP_
// -*-mode:c++; coding:utf-8-*-

#include <iostream>
#include <utility>
#include <memory>
using namespace std;

#include <xbase_exception.hpp>
#include <xbase_datagram_factory.hpp>

ClientMessageHandler::ClientMessageHandler (unsigned int id,
					    handler_status &status)
	: Thread()
	, m_status(status)
	, _id(id)
	, _input_count(0)
	, _output_count(0)
{
}

void ClientMessageHandler::run () {
	LOG(TRACE, "client message handler " << _id << " started.");

	while (true) {
		MessageType msg;
		// get a message
		in_queue.block_pop(msg);
		++ _input_count;
			
		LOG(TRACE, "in hanlder " << *this
		    << " got " << msg.second->remaining() << "bytes message");
		
		// decode it
		int type = XBaseDatagramFactory::getType (*(msg.second));
		switch (type) {
		case READ_REQ:
			this->do_read(msg);
			break;
		case WRITE_REQ:
			this->do_write(msg);
			break;
		case SCAN_OPEN_REQ:
			this->do_scan_open(msg);
			break;
		case SCAN_NEXT_REQ:
			this->do_scan_next(msg);
			break;
		case SCAN_CLOSE_REQ:
			this->do_scan_close(msg);
			break;
		case LIST_TABLET_REQ:
			this->do_list_tablet(msg);
			break;
		default:
			this->do_unexpected(msg);
			break;
		}
	}

	LOG(INFO, "client message handler " << _id << " quit.");
}

// 
// the following function should never throws
//
// TODO: 现在出错后只简单的将连接关闭，以后可能做的完善一些：解码失败的
// 将连接关闭；解码成功的返回失败
// 

// 调试模式下，不捕获异常，以便暴露问题。
#ifndef NDEBUG
#	define try if(true)
#	define catch(x) else
#endif	// NDEBUG

void ClientMessageHandler::do_read(const MessageType &p_msg)
{
	LOG(TRACE, "ClientMessageHandler::do_read");

	std::auto_ptr<ByteBuffer> _holder(p_msg.second);
	MessageType _resp_msg = p_msg;

	try {
		ReadReq _req;
		XBaseDatagramFactory::decode(*(p_msg.second), _req);

		ReadResp _resp;
	
		//boost::shared_ptr<ByteBuffer> _buffer(new ByteBuffer(MAX_PACKET_SIZE));
		//_resp_msg = p_msg;//(p_msg.first, _buffer);
		_resp_msg.second->clear();
		XBaseDatagramFactory::encode(*_resp_msg.second, _resp);

		out_queue.block_push(_resp_msg);
		_holder.release();
	} catch(.../*const std::exception &e*/) {
		LOG(ERROR, "ClientMessageHandler::do_read throw exception of "
		    // << typeid(e).name()
		    << "..., close this socket");
		// 通过发送NULL包，关闭这个连接
		MessageType close_resp(p_msg.first, 0);
		out_queue.block_push(close_resp);
	}
}

void ClientMessageHandler::do_write(const MessageType &p_msg)
{
	LOG(TRACE, "ClientMessageHandler::do_write");

	std::auto_ptr<ByteBuffer> _holder(p_msg.second);
	MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);

	try {
		// WriteReq _req;
		// XBaseDatagramFactory::decode(*(p_msg.second), _req);

		//const write_request_wrapper *req = reinterpret_cast<const write_request_wrapper*>(p_msg.second->array());
		
		WriteResp _resp;
		
		//boost::shared_ptr<ByteBuffer> _buffer(new ByteBuffer(MAX_PACKET_SIZE));
		//MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);
		_resp_msg.second->clear();
		XBaseDatagramFactory::encode(*_resp_msg.second, _resp);
	
		out_queue.block_push(_resp_msg);
		_holder.release();
	} catch(.../*const std::exception &e*/) {
		LOG(ERROR, "ClientMessageHandler::do_write throw exception of "
		    // << typeid(e).name()
		    << "..., close this socket");
		// 通过发送NULL包，关闭这个连接
		MessageType close_resp(p_msg.first, 0);
		out_queue.block_push(close_resp);
	}
}

void ClientMessageHandler::do_scan_open(const MessageType &p_msg)
{
	LOG(TRACE, "ClientMessageHandler::do_scan_open");

	std::auto_ptr<ByteBuffer> _holder(p_msg.second);
	MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);

	try {
		ScanOpenReq _req;
		XBaseDatagramFactory::decode(*(p_msg.second), _req);

		LOG(TRACE, "ClientMessageHandler::do_scan_open req: " << _req.to_string());
		
		ScanOpenResp _resp;
		
		//boost::shared_ptr<ByteBuffer> _buffer(new ByteBuffer(MAX_PACKET_SIZE));
		//MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);
		_resp_msg.second->clear();
		XBaseDatagramFactory::encode(*_resp_msg.second, _resp);
	
		out_queue.block_push(_resp_msg);
		_holder.release();
	} catch(.../*const std::exception &e*/) {
		LOG(ERROR, "ClientMessageHandler::do_scan_open throw exception of "
		    // << typeid(e).name()
		    << "..., close this socket");
		// 通过发送NULL包，关闭这个连接
		MessageType close_resp(p_msg.first, 0);
		out_queue.block_push(close_resp);
	}
}

void ClientMessageHandler::do_scan_next(const MessageType &p_msg)
{
	LOG(TRACE, "ClientMessageHandler::do_scan_next");

	std::auto_ptr<ByteBuffer> _holder(p_msg.second);
	MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);
	
	try {
		ScanNextReq _req;
		XBaseDatagramFactory::decode(*(p_msg.second), _req);

		ScanNextResp _resp;

		//boost::shared_ptr<ByteBuffer> _buffer(new ByteBuffer(MAX_PACKET_SIZE));
		//MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);
		_resp_msg.second->clear();
		XBaseDatagramFactory::encode(*_resp_msg.second, _resp);

		out_queue.block_push(_resp_msg);
		_holder.release();
	} catch(.../*const std::exception &e*/) {
		LOG(ERROR, "ClientMessageHandler::do_scan_next throw exception of "
		    //<< typeid(e).name()
		    << "..., close this socket");
		// 通过发送NULL包，关闭这个连接
		MessageType close_resp(p_msg.first, 0);
		out_queue.block_push(close_resp);
	}
}

void ClientMessageHandler::do_scan_close(const MessageType &p_msg)
{
	TRACE("ClientMessageHandler::do_scan_close");
	TIME_EVALUATOR("ClientMessageHandler::do_scan_close ", "used ");

	std::auto_ptr<ByteBuffer> _holder(p_msg.second);
	MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);

	try {
		////ScanCloseReq _req;
		////XBaseDatagramFactory::decode(*(p_msg.second), _req);
		
		ScanCloseResp _resp;
		
		//boost::shared_ptr<ByteBuffer> _buffer(new ByteBuffer(MAX_PACKET_SIZE));
		//MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);
		_resp_msg.second->clear();
		XBaseDatagramFactory::encode(*_resp_msg.second, _resp);

		_resp_msg.second->flip();
		p_msg.first->write(*_resp_msg.second);

		//out_queue.block_push(_resp_msg);
		//_holder.release();
	} catch(.../*const std::exception &e*/) {
		LOG(ERROR, "ClientMessageHandler::do_scan_close throw exception of "
		    // << typeid(e).name()
		    << "..., close this socket");
		// 通过发送NULL包，关闭这个连接
		MessageType close_resp(p_msg.first, 0);
		out_queue.block_push(close_resp);
	}
}

void ClientMessageHandler::do_unexpected(const MessageType &p_msg)
{
	LOG(WARN, "ClientMessageHandler::do_unexpected unexpected message type 0x"
	    << std::hex << std::uppercase
	    << XBaseDatagramFactory::getType(*(p_msg.second))
	    << ".");

	delete p_msg.second;

	// 通过发送NULL包，关闭这个连接
	MessageType close_resp(p_msg.first, 0);
	out_queue.block_push(close_resp);
}

void ClientMessageHandler::do_list_tablet(const MessageType &p_msg)
{
	LOG(TRACE, "ClientMessageHandler::do_list_tablet");

	std::auto_ptr<ByteBuffer> _holder(p_msg.second);
	MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);

	try {
		list_tablet_resp _resp;
		
		//boost::shared_ptr<ByteBuffer> _buffer(new ByteBuffer(MAX_PACKET_SIZE));
		//MessageType _resp_msg = p_msg;//(p_msg.first, _buffer);
		_resp_msg.second->clear();
		XBaseDatagramFactory::encode(*_resp_msg.second, _resp);

		out_queue.block_push(_resp_msg);
		_holder.release();
	} catch(.../*const std::exception &e*/) {
		LOG(ERROR, "ClientMessageHandler::do_list_tablet throw exception of "
		    // << typeid(e).name()
		    << "..., close this socket");
		// 通过发送NULL包，关闭这个连接
		MessageType close_resp(p_msg.first, 0);
		out_queue.block_push(close_resp);
	}
}
// -*-c++-*-


#ifndef _CLIENT_MESSAGE_PROCESSOR_HPP_
#define _CLIENT_MESSAGE_PROCESSOR_HPP_

#include <stdint.h>
#include <ext/hash_map>
#include <ext/hash_set>
#include <utility>

#include <xsocket/imessageprocessor.hpp>
#include <xsocket/connectioninfo.hpp>

#include <xbase_datagram_factory.hpp>
#include <xbase_utility.hpp>
#include <stringable.hpp>

class ClientMessageProcessor : public IMessageProcessor
{

public:
	// typedefs
	typedef __gnu_cxx::hash_map<uint32_t, ConnectionInfo* > ConnectionMap;
	typedef __gnu_cxx::hash_set<ConnectionInfo* > ConnectionSet;

public:

	ClientMessageProcessor ();

	virtual ~ClientMessageProcessor ();

	virtual void accept (ConnectionInfo &conn);
	virtual void remove (ConnectionInfo &conn);

	virtual void process_input (ConnectionInfo &conn, ByteBuffer &bb);
	virtual void process_output ();

	virtual int frame (ByteBuffer &bb);

	// for debug
	std::string to_string () const {
		std::stringstream out;
		out << "{handlerNumber:" << _handlerNumber
		    << "; connections_size:" << _connections.size ()
		    << "; connections_by_id_size:" << _connections_by_id.size ()
		    << "}";
		
		return out.str();
	}

private:
	// TODO: 用线程池来实现多个Handler
	
	// 客户端请求的处理线程组
	unsigned int _handlerNumber;
	ClientMessageHandler **_handlers;
	handler_status *m_handler_status;
	unsigned int m_cur_handler;

	ConnectionSet _connections;
	ConnectionMap _connections_by_id;
	// unsigned int _max_connetions;

	// for debug & profile
	unsigned int _input_count;
	unsigned int _sent_input_count;
	unsigned int _output_count;
	unsigned int _sent_output_count;
};

#endif	// _CLIENT_MESSAGE_PROCESSOR_HPP_
// -*-mode:c++; coding:utf-8-*-

#include <exception>
#include <utility>
#include <algorithm>
using namespace std;

#include <boost/foreach.hpp>

#include <xsocket/bytebuffer.hpp>

#include <xbase_exception.hpp>
#include <xbase_utility.hpp>
#include <logger.hpp>

ClientMessageProcessor::ClientMessageProcessor ()
	: _handlerNumber(10)
	, _handlers(NULL)
	, m_handler_status(0)
	, m_cur_handler(0)
	, _input_count(0)
	, _output_count(0)
{
	_handlers = new ClientMessageHandler* [_handlerNumber];
	m_handler_status = new handler_status[_handlerNumber];

	// 初始化
	for (unsigned int i=0; i<_handlerNumber; ++i)
	{
		_handlers [i] = NULL;
	}
	for (unsigned int i=0; i<_handlerNumber; ++i)
	{
		_handlers [i] = new ClientMessageHandler(i, m_handler_status[i]);
		_handlers [i]->start();
	}
	LOG(TRACE, _handlerNumber << " client message handlers started.");
}


ClientMessageProcessor::~ClientMessageProcessor () {
	if (_handlers != NULL)
	{
		for (unsigned int i=0; i<_handlerNumber; ++i)
		{
			if (_handlers [i] != NULL)
			{
				delete _handlers [i];
				_handlers [i] = NULL;
			}
		}
		
		delete [] _handlers;
		_handlers = NULL;
	}

	// first save connection in list
	std::vector<ConnectionInfo*> _list;
	BOOST_FOREACH(ConnectionInfo *_p, _connections)
	{
		_list.push_back(_p);
	}
	// then close all alive connection
	BOOST_FOREACH(ConnectionInfo *_p, _list)
	{
		_p->close();
	}
}

void ClientMessageProcessor::process_input (ConnectionInfo &conn, ByteBuffer &bb) {
	TRACE("ClientMessageProcessor::process_input");
	TIME_EVALUATOR("ClientMessageProcessor::process_input ", " used ");

	// update load info

	// for debug
	++ _input_count;

	//int conn_id = conn.get_id ();

	// 拷贝到新的bytebuffer
	//boost::shared_ptr<ByteBuffer> buffer(new ByteBuffer(bb.remaining()));
	std::auto_ptr<ByteBuffer> buffer(new ByteBuffer(std::max((int)MAX_PACKET_SIZE, bb.remaining())));
	buffer->put(bb);
	buffer->flip();
	// ByteBuffer::arraycopy (bb.array (), bb.position (),
	// 		       buffer->array (), buffer->position (),
	// 		       bb.remaining ());
	// buffer->limit(bb.remaining ());
	
	MessageType msg(&conn, buffer.get());

	// 选择一个处理线程进行处理
	// TODO: 用更好的选择策略，当前的策略是同一个Socket的数据被放到相同的处理线程
	//_handlers [conn_id % _handlerNumber]->sendMessage (msg);

	in_queue.block_push(msg);

	buffer.release();
	
	++m_cur_handler;
	if(m_cur_handler >= _handlerNumber) {
		m_cur_handler = 0;
	}

	// for debug
	++ _sent_input_count;
	
}

void ClientMessageProcessor::process_output () {
	// for load info
	// unsigned int left_in_message = 0;
	// unsigned int left_out_message = 0;

	// //MessageType msg;
	// unsigned int count_per_handler = 0;

	// for (unsigned int i=0; i<_handlerNumber; ++i) {
	// 	count_per_handler = 0;
	// 	for (int j=0; j<128; ++j) {
	// 		MessageType msg;
	// 		if(! out_queue.pop(msg)) // empty
	// 			break;

	// 		TRACE("ClientMessageProcessor::process_output");
	// 		TIME_EVALUATOR("ClientMessageProcessor::process_output ", " used ");
	
	// 		// for debug
	// 		// 记录成功取出及总的的响应包数，用于统计。
	// 		++ count_per_handler;
	// 		++ _output_count;
			
	// 		boost::scoped_ptr<ByteBuffer> buffer(msg.second);
	// 		// 准备发送数据包，打印日志也需要先 flip
	// 		if (buffer)
	// 			buffer->flip ();
			
	// 		ConnectionMap::iterator iter = _connections_by_id.find (msg.first);
			
	// 		// 没有该连接，可能是处理期间已经关闭或
	// 		// 未知的连接，直接将其丢弃。
	// 		if (iter == _connections_by_id.end()) {
	// 			ostringstream stream;
	// 			if (buffer)
	// 				stream << CommandName(XBaseDatagramFactory::getType(*buffer));
	// 			else
	// 				stream << "NULL";
				
	// 			LOG(ERROR, "response message("
	// 			    << stream.str()
	// 			    << ") to client on a"
	// 			    << " CLOSED or UNKNOWN connection, drop it.");
				
	// 			continue;
	// 		}
			
	// 		// 通常情况下，连接断开就会从列表中删除
	// 		// 的，所以这里取到的一定是存活的连接，
	// 		// 但以防未知的错误发生，这里进行了一次
	// 		// 检查
			
	// 		// 若是已关闭连接，则不应该出现在该列表
	// 		// 中，检测出来后将其删除，并将相关的响
	// 		// 应包丢弃
	// 		if(! iter->second->is_open()) {
	// 			ostringstream stream;
	// 			if (buffer)
	// 				stream << CommandName(XBaseDatagramFactory::getType(*buffer));
	// 			else
	// 					stream << "NULL";
				
	// 			LOG(ERROR, "response message("
	// 			    << stream.str()
	// 				    <<") to client on a"
	// 			    << " CLOSED connection, drop it.");
				
	// 			this->remove(*iter->second); // must first
	// 			iter->second->close(); // must second
	// 			continue;
	// 		}
			
	// 		// 空的数据指针，则将连接关闭
	// 		if (! buffer) {
	// 			LOG(INFO, "got a NULL response buffer on connection "
	// 			    << *iter->second << ", so close this connection.");

	// 			// 前面已经保证该连接存在并没有
	// 			// 关闭，因此只调用close即可
	// 			iter->second->close();
	// 			continue;
	// 		}
			
	// 		// 若是一个空的数据包，也将连接关闭
	// 		if (buffer->remaining () == 0) {
	// 			LOG(INFO, "got a EMPTY response buffer, so close the connection.");
				
	// 			iter->second->close();
	// 			continue;
	// 		}
			
	// 		// 若没准备好写，则可能是上次的没有发送
	// 		// 完，或连接出问题了，尝试将其flush一
	// 		// 下再作处理
	// 		if(! iter->second->ready_to_write()) {
	// 			LOG(TRACE, "flush on connection " << *iter->second << ".");
	// 			iter->second->flush ();
	// 		}
			
	// 		// 若仍然不能写，很可能是出问题了，也将
	// 		// 其关闭。
	// 		if(! iter->second->ready_to_write()) {
	// 			LOG(ERROR, "connection " << *iter->second
	// 			    << " not ready to write, so close it."
	// 			    << "(should me?)");
				
	// 			iter->second->close ();
	// 			continue;
	// 		}

	// 		// 到现在才排除了各种可能的问题和错误，开始发送包
	// 		int ret = iter->second->write (*buffer);
			
	// 		// 若没发送成功，则将连接关闭
	// 		if(ret < 0) {
	// 			LOG(ERROR, "write to connection "
	// 			    << *iter->second
	// 			    << " failed, so close it.");
				
	// 			// 关闭操作会调用自身的remove函
	// 			// 数将其删除
	// 			iter->second->close ();
	// 			continue;
	// 		}
			
	// 		// 到此才发送成功，设置统计用的计数器
	// 		++ _sent_output_count;
	// 	}
		
	// 	// get load info
	// 	left_in_message += m_handler_status[i].m_left_in_message;
	// 	left_out_message += m_handler_status[i].m_left_out_message;
	// }
}

void ClientMessageProcessor::accept (ConnectionInfo &conn) {
	// DONE: 这里不需要考虑插入失败吗？虽然基本上不会有两个相同的
	// Key，除非是同一个对象。而这时候它已经在里面了，就不用再插入
	// 了吧。
	_connections.insert (&conn);
	_connections_by_id.insert (ConnectionMap::value_type(conn.get_id(), &conn));

	LOG(INFO, "a connection from " << conn << " accepted.");
}

void ClientMessageProcessor::remove (ConnectionInfo &conn) {
	_connections.erase (&conn);
	_connections_by_id.erase (conn.get_id());

	LOG(INFO, "a connection from " << conn << " closed.");
}

int ClientMessageProcessor::frame (ByteBuffer &bb) {
	return XBaseDatagramFactory::frame (bb);
}

int main(int argc, char **argv)
{
	if(argc != 2)
	{
		std::cout << argv[0] << " <select_timeout>" << std::endl;
		exit(-1);
	}

	const int timeout = std::atoi(argv[1]);

	ClientMessageProcessor processor;
	ActiveReactor reactor(timeout);
	reactor.init();
	const int ret = reactor.add_acceptor("0.0.0.0", 20083, processor);
	if(ret < 0)
	{
		return -1;
	}

	while(true)
	{
		reactor.poll();

		// wasting time
		int n=9;
		time_t time = cur_micro_seconds();
		while(n--)
		{
			time += cur_micro_seconds();
		}
	}

	return 0;
}
